package com.shujia.flink.source

import org.apache.flink.streaming.api.functions.source.SourceFunction

class Demo2TesSource extends SourceFunction[Int] {

  /**
    * 生产数据  ，
    * 连接外部数据库   外部系统获取数据
    *
    * run 方法在一个task中只会调用异常
    *
    */

  override def run(ctx: SourceFunction.SourceContext[Int]): Unit = {

    var i: Int = 0

    //不停的生产数据
    while (true) {
      i += 1

      // 将数据发送到下游
      ctx.collect(i)


      Thread.sleep(1000)
    }

  }

  /**
    * 回收资源
    *
    */

  override def cancel(): Unit = {

  }
}
